package com.mai.realtime.function;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public abstract class chSinkFunction<T> extends RichSinkFunction<T>{

    private Connection conn;
    private PreparedStatement insertStmt;
    @Override
    public void open(Configuration parameters) throws Exception {
        conn = DriverManager.getConnection(
                "jdbc:clickhouse://hadoop162:8123/default"
        );
    }


    @Override
    public void close() throws Exception {

        insertStmt.close();
        conn.close();
    }



    @Override
    public void invoke(T value, Context ctx) throws Exception {
        String sql=getSql();
        insertStmt = conn.prepareStatement(sql);
        addPs(insertStmt,value);
        insertStmt.execute();

    }

    protected abstract String getSql();

    protected abstract void addPs(PreparedStatement ps, T value);



}